package mork;

import bean.Case.Message;
import bean.Case.category;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @program: mork
 * @author: YCLW058
 * @create: 2021-05-26 15:52
 * @decsription:
 **/

public class catergoryPojoMork {
    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        //kafka配置
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        //数据写入kafka的 模式 0 表示 发送即可 1 表示 磁盘写入即可  all 表示 写入后 需要返回是否成功
        props.put("acks", "all");
        //重试连接次数
        props.put("retries", 2);
        //重试连接时间
        props.put("retries.backoff.ms", 20);
        //缓存大小
        props.put("buffer.memory", 33554432);
        //批次数据大小
        props.put("batch.size", 16384);
        //批次时间   数据只要达到 批次数据大小或者 批次时间 数据就会发送出去
        props.put("linger.ms", 25);
        //最大发送数据
        props.put("max.request.size", 163840);
        //key 序列化
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value 序列化
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> kafkaproducer = new KafkaProducer<>(props);

        Random random = new Random();
        Boolean flag = true;

        // {"count":1,"message":"{\"buyTime\":1577880060000,\"buyType\":\"361\",\"categoryID\":17,\"country\":\"china\",\"leaveTime\":1577880060000,\"network\":\"电信\",\"price\":59801.538972401155,\"produceID\":19,\"source\":\"实体店\",\"userID\":14}","timeStamp":1622095734031}

        while (flag) {
            //构建数据
            category categorylog = new category();
            categorylog.setCategoryID(categoryIDs[random.nextInt(categoryIDs.length)]);
            categorylog.setProduceID(produceIDs[random.nextInt(produceIDs.length)]);
            categorylog.setUserID(userIDs[random.nextInt(userIDs.length)]);
            categorylog.setPrice(random.nextDouble() * 100000);
            categorylog.setCountry(contrys[random.nextInt(contrys.length)]);
            categorylog.setNetwork(networks[random.nextInt(networks.length)]);
            categorylog.setSource(sources[random.nextInt(sources.length)]);
            categorylog.setBuyType(buyTypes[random.nextInt(buyTypes.length)]);
            Long[] times = usetimelog.get(random.nextInt(usetimelog.size()));
            categorylog.setBuyTime(times[0]);
            categorylog.setLeaveTime(times[0]);
            String logJson = JSONObject.toJSONString(categorylog);

            //构建msg
            Message msg = new Message();
            msg.setMessage(logJson);
            msg.setCount(1);
            msg.setTimeStamp(System.currentTimeMillis());
            String msgJson = JSON.toJSONString(msg);


            //发送数据
            //消费者
            ProducerRecord<String, String> input_kafka = new ProducerRecord<>("input_kafka", msgJson);
            kafkaproducer.send(input_kafka);

            System.out.println("数据已经发送到kafka："+msgJson);
            Thread.sleep(500);

        }
        kafkaproducer.close();
    }

    //产品类别id集合
    private static Long[] categoryIDs = new Long[]{1l, 2l, 3l, 4l, 5l, 6l, 7l, 8l, 9l, 10l, 11l, 12l, 13l, 14l, 15l, 16l, 17l, 18l, 19l, 20l};
    //产品id集合
    private static Long[] produceIDs = new Long[]{1l, 2l, 3l, 4l, 5l, 6l, 7l, 8l, 9l, 10l, 11l, 12l, 13l, 14l, 15l, 16l, 17l, 18l, 19l, 20l};
    //用户id集合
    private static Long[] userIDs = new Long[]{1l, 2l, 3l, 4l, 5l, 6l, 7l, 8l, 9l, 10l, 11l, 12l, 13l, 14l, 15l, 16l, 17l, 18l, 19l, 20l};
    //地区-国家集合
    private static String[] contrys = new String[]{"china"};
    //网络方式集合
    private static String[] networks = new String[]{"电信", "移动", "联通"};
    //来源方式
    private static String[] sources = new String[]{"官网", "第三方", "实体店", "百度链接"};
    //购买类型集合
    private static String[] buyTypes = new String[]{"淘宝", "京东", "得物", "安踏", "361"};

    //获取时间
    private static List<Long[]> usetimelog = producetimes();

    public static List<Long[]> producetimes() {
        List<Long[]> usetimelog = new ArrayList<Long[]>();
        for (int i = 0; i < 100; i++) {
            Long[] timesarray = gettimes("2020-01-01 24:60:60:000");
            usetimelog.add(timesarray);
        }
        return usetimelog;
    }

    private static Long[] gettimes(String time) {
        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss:SSS");
        try {
            Date date = dateFormat.parse(time);
            long timetemp = date.getTime();
            Random random = new Random();
            int randomint = random.nextInt(10);
            long starttime = timetemp - randomint * 3600 * 1000;
            long endtime = starttime + randomint * 3600 * 1000;
            return new Long[]{starttime, endtime};
        } catch (ParseException e) {
            e.printStackTrace();
        }
        return new Long[]{0L, 0L};
    }
}
